/**
 * Date:2021年9月18日 下午3:46:40
 * Copyright (c) 2021, www.bwbroad.com All Rights Reserved
 */

package com.xue;

import java.io.IOException;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;

/**
 * Description: TODO <br/>
 * Date:2021年9月18日 下午3:46:40 <br/>
 * 
 * @author xuejianxin
 */
public class Debezium {
	public static void config() throws IOException {
		final Properties props = new Properties();
		props.setProperty("name", "engine");
		props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
		props.setProperty("offset.storage.file.filename", "offsets.dat");
		props.setProperty("offset.flush.interval.ms", "5000");
		/* begin connector properties */
		props.setProperty("connector.class", "io.debezium.connector.oracle.OracleConnector");
		props.setProperty("database.server.name", "oracle12cTest");
		props.setProperty("database.hostname", "172.16.10.139");
		props.setProperty("database.port", "31521");
		props.setProperty("database.dbname", "orcl");
		props.setProperty("database.user", "dbzuser");
		props.setProperty("database.password", "dbz");
		props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
		props.setProperty("database.history.file.filename", "dbhistory.dat");
		//props.setProperty("converter.schemas.enable", "false"); // don't include schema in message
		//props.setProperty("schema.include.list", "apex_040200");
		props.setProperty("table.include.list", "apex_040200.customers");
		//props.setProperty("snapshot.mode", "schema_only");
		
		


		// Create the engine with this configuration ...
		try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class).using(props)
				.notifying(record -> {
					System.out.println(record);
				}).build()) {
			// Run the engine asynchronously ...
			ExecutorService executor = Executors.newSingleThreadExecutor();
			executor.execute(engine);

			// Do something else or wait for a signal or an event
			
		}
	}

	
	public static void main(String[] args) throws IOException {
		config();
		System.in.read();
	}
}
